package com.lecosa.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class ChangeSinkJson extends RichSinkFunction<String> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    private String lastValue = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 加载驱动程序
        Class.forName("com.mysql.cj.jdbc.Driver");
        // 获取数据库连接
        String url = "jdbc:mysql://192.168.249.112:3306/shufen";
        String user = "root";
        String password = "root";
        connection = DriverManager.getConnection(url, user, password);
        // 创建 PreparedStatement 对象
        String sql = "replace INTO student1 (id, no, name, gender) VALUES (?, ?, ?, ?)";
        preparedStatement = connection.prepareStatement(sql);
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 将 JSON 字符串解析为 JSONObject
        JSONObject jsonObject = JSON.parseObject(value);

        // 从 JSONObject 中提取 after 字段的值
        JSONObject afterObject = jsonObject.getJSONObject("after");
        int id = afterObject.getInteger("id");
        String no = afterObject.getString("no");
        String name = afterObject.getString("name");
        String gender = afterObject.getString("gender");

        // 如果上一次处理的值为空，或者当前值与上一次处理的值不相同，则执行 sink 操作
        if (lastValue == null || !lastValue.equals(value)) {
            // 将值插入到新表中
            preparedStatement.setInt(1, id);
            preparedStatement.setString(2, no);
            preparedStatement.setString(3, name);
            preparedStatement.setString(4, gender);
            preparedStatement.executeUpdate();

            // 更新上一次处理的值为当前值
            lastValue = value;
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接和 PreparedStatement
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
